package com.bw.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.MyKafkaUtil;
import com.bw.bean.DwdBoundSortBean;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class Test5 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        env.setParallelism(1);

        DataStreamSource<String> stream = env.addSource(MyKafkaUtil.getKafkaConsumer("tms_ods_yk9","test15"));

        //筛选订单和订单明细数据
        SingleOutputStreamOperator<String> DwdBoundSortBeanDS = stream.filter(x -> JSON.parseObject(x).getString("table").equals("order_org_bound"));
        //对流中的数据类型进行转换

        SingleOutputStreamOperator<JSONObject> etlStream = DwdBoundSortBeanDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                System.out.println("s = " + s);
                try {
                    // 先转成JSON
                    JSONObject jsonObject = JSONObject.parseObject(s);
                    // 拿到操作 insert update delete
                    String op = jsonObject.getString("op");
                    if (op.equals("c")) {
                        collector.collect(jsonObject);
                    } else {
                        JSONObject dataJsonObject = jsonObject.getJSONObject("data");
                        if (dataJsonObject != null && dataJsonObject.getString("sort_time") != null) {
                            collector.collect(jsonObject);
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        //64002
        OutputTag<DwdBoundSortBean> t64002 = new OutputTag<DwdBoundSortBean>("t64002"){};
        //64003
        OutputTag<DwdBoundSortBean> t64003 = new OutputTag<DwdBoundSortBean>("t64003"){};
        //64004
        OutputTag<DwdBoundSortBean> t64004 = new OutputTag<DwdBoundSortBean>("t64004"){};
        //64005
        OutputTag<DwdBoundSortBean> t64005 = new OutputTag<DwdBoundSortBean>("t64005"){};

        // 分流
        SingleOutputStreamOperator<DwdBoundSortBean> splitStream = etlStream.process(new ProcessFunction<JSONObject, DwdBoundSortBean>() {
            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, DwdBoundSortBean>.Context context, Collector<DwdBoundSortBean> collector) throws Exception {
                JSONObject data = jsonObject.getJSONObject("data");
                String status = data.getString("status");
                DwdBoundSortBean bean = JSON.parseObject(data.toJSONString(), DwdBoundSortBean.class);
                if ("64002".equals(status)) {
                    context.output(t64002, bean);
                } else if ("64003".equals(status)) {
                    context.output(t64003, bean);
                } else if ("64004".equals(status)) {
                    context.output(t64004, bean);
                } else if ("64005".equals(status)) {
                    context.output(t64005, bean);
                } else {
                    collector.collect(bean);
                }
            }
        });

        // 提流
        splitStream.print();
        // 把对象转成JSON
        splitStream.map(JSON::toJSONString).addSink(MyKafkaUtil.getKafkaProducer("dwd_order_org_bound_p64001_yk9"));
        splitStream.getSideOutput(t64002).map(JSON::toJSONString).addSink(MyKafkaUtil.getKafkaProducer("dwd_order_org_bound_p64002_yk9"));
        splitStream.getSideOutput(t64003).map(JSON::toJSONString).addSink(MyKafkaUtil.getKafkaProducer("dwd_order_org_bound_p64003_yk9"));
        splitStream.getSideOutput(t64004).map(JSON::toJSONString).addSink(MyKafkaUtil.getKafkaProducer("dwd_order_org_bound_p64004_yk9"));
        splitStream.getSideOutput(t64005).map(JSON::toJSONString).addSink(MyKafkaUtil.getKafkaProducer("dwd_order_org_bound_p64005_yk9"));

        env.execute();
    }
}
